package etcd

import (
	"context"
	"errors"
	"time"

	"gark-elastic-job/engine"
	"github.com/sirupsen/logrus"
	"go.etcd.io/etcd/clientv3"
)

type Etcd struct {
	endpoints []string
	client    *clientv3.Client
	kv        clientv3.KV
	timeout   time.Duration
}

func NewEtcd(endpoints []string, timeout time.Duration) (*Etcd, error) {
	client, err := clientv3.New(clientv3.Config{
		Endpoints:   endpoints,
		DialTimeout: timeout,
	})

	if err != nil {
		return nil, err
	}

	return &Etcd{
		endpoints: endpoints,
		client:    client,
		kv:        clientv3.NewKV(client),
		timeout:   timeout,
	}, nil
}

func (etcd Etcd) Get(key string) ([]byte, error) {
	ctx, cancelFunc := context.WithTimeout(context.Background(), etcd.timeout)
	defer cancelFunc()

	getResponse, err := etcd.client.Get(ctx, key)
	if err != nil {
		return nil, err
	}

	if len(getResponse.Kvs) == 0 {
		return nil, errors.New("key[ " + key + " ] does not exist")
	}

	return getResponse.Kvs[0].Value, nil
}

func (etcd Etcd) Put(key, value string) error {
	ctx, cancelFunc := context.WithTimeout(context.Background(), etcd.timeout)
	defer cancelFunc()

	_, err := etcd.client.Put(ctx, key, value)
	if err != nil {
		return err
	}

	return nil
}

// PutNotExist put the value into the key
//	if the key already exists return false and the previous-value
//  if the key not exist return true
func (etcd Etcd) PutNotExist(key, value string) (bool, []byte, error) {
	ctx, cancelFunc := context.WithTimeout(context.Background(), etcd.timeout)
	defer cancelFunc()

	txn := etcd.client.Txn(ctx)
	txnResponse, err := txn.If(clientv3.Compare(clientv3.Version(key), "=", 0)).
		Then(clientv3.OpPut(key, value)).
		Else(clientv3.OpGet(key)).
		Commit()
	if err != nil {
		return false, nil, err
	}

	if txnResponse.Succeeded {
		return true, nil, nil
	} else {
		prevValue := txnResponse.Responses[0].GetResponseRange().Kvs[0].Value
		return false, prevValue, nil
	}
}

func (etcd Etcd) GetWithPrefixKey(prefixKey string) ([][]byte, [][]byte, error) {
	var (
		keyResp   [][]byte
		valueResp [][]byte
	)

	ctx, cancelFunc := context.WithTimeout(context.Background(), etcd.timeout)
	defer cancelFunc()
	getResp, err := etcd.client.Get(ctx, prefixKey, clientv3.WithPrefix())

	if err != nil {
		return nil, nil, err
	}
	for _, kv := range getResp.Kvs {
		keyResp = append(keyResp, kv.Key)
		valueResp = append(valueResp, kv.Value)
	}
	return keyResp, valueResp, nil
}

func (etcd Etcd) GetWithPrefixKeyLimit(prefixKey string, limit int64) ([][]byte, [][]byte, error) {
	panic("implement me")
}

// TxLeaseTTL ttl -> seconds
func (etcd Etcd) TxLeaseTTL(key, val string, ttl int64) (txResponse *engine.TxResponse, err error) {

	var (
		grantResponse *clientv3.LeaseGrantResponse
		txnResponse   *clientv3.TxnResponse
		v             []byte
	)

	lease := clientv3.NewLease(etcd.client)
	grantResponse, err = lease.Grant(context.Background(), ttl)
	if err != nil {
		return nil, err
	}
	leaseID := grantResponse.ID
	ctx, cancelFunc := context.WithTimeout(context.Background(), etcd.timeout)
	defer cancelFunc()
	txn := etcd.client.Txn(ctx)

	txnResponse, err = txn.
		If(clientv3.Compare(clientv3.Version(key), "=", 0)).
		Then(clientv3.OpPut(key, val, clientv3.WithLease(leaseID))).
		Commit()

	if err != nil {
		_ = lease.Close()
		return
	}

	txResponse = &engine.TxResponse{
		LeaseID: leaseID,
		Lease:   lease,
	}

	if txnResponse.Succeeded {
		txResponse.Success = true
	} else {
		_ = lease.Close()
		v, err = etcd.Get(key)
		if err != nil {
			return
		}
		txResponse.Success = false
		txResponse.Key = key
		txResponse.Value = string(v)
	}
	return
}

// TxKeepAliveLeaseTTL lease per 10 second
func (etcd Etcd) TxKeepAliveLeaseTTL(key, val string, ttl int64) (txResponse *engine.TxResponse, err error) {

	var (
		grantResp         *clientv3.LeaseGrantResponse
		txnResp           *clientv3.TxnResponse
		v                 []byte
		keepAliveRespChan <-chan *clientv3.LeaseKeepAliveResponse
	)

	lease := clientv3.NewLease(etcd.client)
	grantResp, err = lease.Grant(context.Background(), ttl)
	if err != nil {
		return
	}

	leaseID := grantResp.ID

	keepAliveRespChan, err = lease.KeepAlive(context.Background(), leaseID)
	if err != nil {
		return
	}

	go func() {
		for ch := range keepAliveRespChan {
			if ch == nil {
				goto END
			}
		}
	END:
		logrus.Infof("the tx keepalive has lose key:%s", key)
	}()

	ctx, cancelFunc := context.WithTimeout(context.Background(), etcd.timeout)
	defer cancelFunc()
	txn := etcd.client.Txn(ctx)

	txnResp, err = txn.
		If(clientv3.Compare(clientv3.Version(key), "=", 0)).
		Then(clientv3.OpPut(key, val, clientv3.WithLease(leaseID))).
		Else(clientv3.OpGet(key)).
		Commit()

	if err != nil {
		_ = lease.Close()
		return
	}

	txResponse = &engine.TxResponse{
		LeaseID: leaseID,
		Lease:   lease,
	}

	if txnResp.Succeeded {
		txResponse.Success = true
	} else {
		_ = lease.Close()
		v, err = etcd.Get(key)
		if err != nil {
			return
		}
		txResponse.Success = false
		txResponse.Key = key
		txResponse.Value = string(v)
	}
	return
}

// 监测变化的key
func (etcd Etcd) WatchKey(key string) (response *engine.WatchKeyChangeResponse) {

	watcher := clientv3.NewWatcher(etcd.client)
	watchChan := watcher.Watch(context.Background(), key)

	response = &engine.WatchKeyChangeResponse{
		Event:   make(chan *engine.KeyChangeEvent, 64),
		Watcher: watcher,
	}

	go func() {
		for ch := range watchChan {
			if ch.Canceled {
				goto END
			}
			for _, event := range ch.Events {
				etcd.handleKeyChangeEvent(response.Event, event)
			}
		}
	END:
		logrus.Info("watcher lost for the key: ", key)
	}()

	return

}

// 监测以某个前缀开头的变化的key
func (etcd Etcd) WatchKeyWithPrefix(prefixKey string) *engine.WatchKeyChangeResponse {

	watcher := clientv3.NewWatcher(etcd.client)
	watchChan := watcher.Watch(context.Background(), prefixKey, clientv3.WithPrefix())

	response := &engine.WatchKeyChangeResponse{
		Event:   make(chan *engine.KeyChangeEvent, 64),
		Watcher: watcher,
	}

	go func() {
		for ch := range watchChan {
			if ch.Canceled {
				goto END
			}
			for _, event := range ch.Events {
				etcd.handleKeyChangeEvent(response.Event, event)
			}
		}
	END:
		logrus.Info("watcher lost for the prefix key ", prefixKey)
	}()

	return response

}

func (etcd Etcd) DeleteKey(key string) error {
	ctx, cancelFunc := context.WithTimeout(context.Background(), etcd.timeout)
	defer cancelFunc()

	_, err := etcd.client.Delete(ctx, key)
	return err
}

func (etcd Etcd) DeleteKeyWithPrefix(key string) error {
	ctx, cancelFunc := context.WithTimeout(context.Background(), etcd.timeout)
	defer cancelFunc()

	_, err := etcd.client.Delete(ctx, key, clientv3.WithPrefix())
	return err
}

func (etcd Etcd) handleKeyChangeEvent(eventChan chan *engine.KeyChangeEvent, event *clientv3.Event) {
	keyChangeEvent := &engine.KeyChangeEvent{
		Key: string(event.Kv.Key),
	}
	switch event.Type {
	case clientv3.EventTypePut:
		if event.IsCreate() {
			keyChangeEvent.Type = engine.KeyCreateChangeEvent
		} else {
			keyChangeEvent.Type = engine.KeyUpdateChangeEvent
		}
		keyChangeEvent.Value = event.Kv.Value
	case clientv3.EventTypeDelete:
		keyChangeEvent.Type = engine.KeyDeleteChangeEvent
	}
	eventChan <- keyChangeEvent
}
